package com.cennavi.funcs;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;

import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月20日 上午9:27:55 
    * @version 1.0 
    */

public class SumFlowPairFunction implements PairFunction<Tuple2<String, Iterable<String>>, String, String> {

	private static final long serialVersionUID = 1L;

	public Tuple2<String, String> call(Tuple2<String, Iterable<String>> t) throws Exception {
		Tuple2<String, String> tupleSumFlow;
		//遍历分组后的每一个key(devid拼接rticid)
		String key = t._1;
		String value = "";
		String[] data = new String[20];
		//合并后的流量值
		int flow_sum = 0;
		//时间差
		long deltaMinutes = 0;
		String minTimestamp = "2200-01-01 00:00:00";
		String maxTimestamp = "1800-01-01 00:00:00";
		int count = 0;
		for(Iterator<String> iter = t._2.iterator(); iter.hasNext();){
			String str = (String)iter.next();
			data = str.split(",");
			flow_sum = flow_sum + Integer.parseInt(data[2]);
			if(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(minTimestamp).getTime() >
				new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(data[3]).getTime()	
					){
				minTimestamp = data[3];
			}
			if(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(maxTimestamp).getTime() <
				new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(data[3]).getTime()	
					){
				maxTimestamp = data[3];
			}
			count ++;
		}
		//分组后大于一行
		if(count > 1){
			deltaMinutes = calMinutes(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(minTimestamp), 
					new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(maxTimestamp));
			//线性拉伸
//			if(deltaMinutes > 0){
//				flow_sum = (int)Math.ceil(1.0 * flow_sum * 15 / deltaMinutes);
//			}
		}else{
			minTimestamp = maxTimestamp = data[3];
		}
		value = String.valueOf(flow_sum) + "," 
				//+ DateUtil.get5min(new Date()) + ","
				+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ","
				+ minTimestamp + ","
				+ maxTimestamp;	
		tupleSumFlow = new Tuple2<String, String>(key, value);
		return tupleSumFlow;
	}

	private long calMinutes(Date date1, Date date2){
		return (date2.getTime() - date1.getTime()) / (1000 * 60);
	}
	
//	public static void main(String[] args) throws ParseException {
//		String timestamp = "2017-09-20 23:46:55";
//		Date date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(timestamp);
//		System.out.println(get15min(date));
//	}
}
